package com.example.webflux.controller;

import cn.hutool.json.JSONUtil;
import com.example.webflux.model.Greeting;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.io.PrintWriter;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@RestController
public class GreetingController {

    @GetMapping("/greeting")
    public Mono<Greeting> greeting(@RequestParam(value = "name", defaultValue = "World") String name) {
        return Mono.just(new Greeting("Hello, " + name));
    }

    // 模拟生成流式数据
    public Flux<String> generateData() {
        return Flux.fromStream(
                IntStream.range(1, 50).mapToObj(i -> "Item-" + i)
        ).concatWithValues(UUID.randomUUID().toString());
    }

    // 示例1：SSE (Server-Sent Events) 流式输出
    @GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamSSE() {
        return generateData()
                .delayElements(Duration.ofSeconds(1))  // 每秒推送一条数据
                .map(data -> "data: " + data + "\n\n");  // SSE格式要求
    }

    // 示例2：JSON Stream 流式输出
    @GetMapping(value = "/json", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
    public Flux<DataItem> streamJson() {
        return generateData()
                .delayElements(Duration.ofMillis(500))  // 每0.5秒推送一个JSON对象
                .map(data -> new DataItem(data,String.valueOf(System.currentTimeMillis())));  // 转换为JSON对象
    }

    // 数据模型（内部类）
    static class DataItem {
        private final String content;
        private final String timestamp;
        public DataItem(String content,String timestamp) { this.content = content; this.timestamp = timestamp;}
        public String getContent() { return content; }
        public String getTimestamp() { return timestamp; }
    }
    // ###########
    @PostMapping("/stream")
    public ResponseEntity<StreamingResponseBody> streamData() {
        Map<String, String> map = new HashMap<>();
        map.put("content", "内容");

        StreamingResponseBody responseBody = outputStream -> {
            try (PrintWriter writer = new PrintWriter(outputStream)) {
                for (int i = 0; i < 10; i++) {
                    map.put("content", "内容:" + i);
                    writer.println(JSONUtil.toJsonStr(map));
                    writer.flush();
                    // 模拟一些延迟
                    Thread.sleep(500);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        HttpHeaders headers = new HttpHeaders();
//        headers.setContentType(MediaType.TEXT_PLAIN);
        headers.add("ContentType","text/plain; charset=UTF-8");
        // 指示这是一个流式响应
        headers.setContentLength(-1L);

        return new ResponseEntity<>(responseBody, headers, HttpStatus.OK);
    }


    @GetMapping("/flux")
    public void test1(){
        String[] arr = {"data1", "data2", "data3", "data4", "data5"};
        // 模拟一些数据源
        List<String> dataSources = Arrays.asList(arr);
        AtomicInteger idx = new AtomicInteger(0);

        // 创建 Flux 和进行操作
        Flux.fromIterable(dataSources)
                .doOnNext(data -> {
                    // 处理每个数据
                    String transformedData = transformData(data);
                    try {
                        // 假设这里发送到某个异步系统
                        sendData(transformedData, idx);
                    } catch (IOException e) {
                        // 错误处理
                        System.err.println("发送失败, 数据: " + transformedData + " 错误信息: " + e.getMessage());
                        // 重试机制
                        retrySend(transformedData, idx).subscribe();
                    }
                })
                .doOnError(err -> {
                    // 处理错误并记录
                    System.err.println("遇到错误: " + err.getMessage());
                })
                .doOnComplete(() -> {
                    // 完成后的回调
                    System.out.println("处理完成！");
                })
                .buffer(3) // 每批处理 3 条数据
                .flatMap(batch -> {
                    // 批处理后再异步处理
                    return Mono.just(batch)
                            .flatMap(b -> processBatchAsync(b));
                })
                .subscribe(
                        result -> System.out.println("异步处理结果: " + result),
                        error -> System.err.println("最终错误: " + error.getMessage())
                );
    }

//    public static void main(String[] args) {
//
//    }

    // 模拟数据转换
    private static String transformData(String data) {
        return data.toUpperCase();
    }

    // 模拟数据发送
    private static void sendData(String data, AtomicInteger idx) throws IOException {
        if (Math.random() > 0.95) { // 随机失败
            throw new IOException("发送失败");
        }
        System.out.println("发送数据成功, 数据: " + data + ", ID: " + idx.incrementAndGet());
    }

    // 重试机制
    private static Mono<Object> retrySend(String data, AtomicInteger idx) {
        return Mono.defer(() -> {
            try {
                sendData(data, idx);
                return Mono.empty();
            } catch (IOException e) {
                return Mono.error(e);
            }
        }).retry(3)  // 最大重试次数 3 次
                .doOnTerminate(() -> System.out.println("重试结束"));
    }

    // 模拟异步批量处理
    private static Mono<String> processBatchAsync(List<String> batch) {
        return Mono.fromCallable(() -> {
            // 模拟一些耗时操作
            Thread.sleep(1000);
            return batch.stream().collect(Collectors.joining(", "));
        });
    }
}
